package com.cw.realtime.common
package util

import constant.Constant._

import com.ververica.cdc.connectors.mysql.source.MySqlSource
import com.ververica.cdc.connectors.mysql.table.StartupOptions
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer

import java.nio.charset.StandardCharsets
import java.util.Properties

object FlinkSourceUtil {
  private class SimpleStringSchemaPatched extends SimpleStringSchema {

    override def  deserialize(message: Array[Byte]): String = {
      // flink sql撤回流会产生null，所以要检查
      if (message == null || message.isEmpty)
        "-D"
      else
        new String(message, StandardCharsets.UTF_8)
    }

  }


  def getKafkaSource(topic: String, groupId: String): KafkaSource[String] = {
    KafkaSource.builder()
      .setTopics(topic)
      .setBootstrapServers(KAFKA_BROKERS)
      .setStartingOffsets(OffsetsInitializer.earliest())
      .setGroupId(groupId)
      .setValueOnlyDeserializer(new SimpleStringSchemaPatched)
      .build()
  }

  def getMysqlSource(database: String, table: String): MySqlSource[String] = {

    val prop =  new Properties
    prop.setProperty("useSSL", "false")
    prop.setProperty("allowPublicKeyRetrieval", "true")
    MySqlSource.builder()
      .hostname(MYSQL_HOST)
      .port(MYSQL_PORT)
      .username(MYSQL_USER_NAME)
      .password(MYSQL_PASSWORD)
      .databaseList(database)
      .tableList(s"${database}.${table}")
      .deserializer(new JsonDebeziumDeserializationSchema())
      .startupOptions(StartupOptions.initial())
      .jdbcProperties(prop)
      .build()
  }


}
